package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MykafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName DwdTradeCardAdd
 * @Description TODO 加购事务事实表
 * @Author AFei
 * @Date 2022/9/6 0:13
 * @Version 1.0
 */
public class DwdTradeCartAdd {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 从topic_db主题中读取业务表变化数据  创建动态表
        tableEnv.executeSql(MykafkaUtil.getTopicDDL("dwd_trade_cart_add"));

        // TODO 从业务数据中筛选出加购数据
        Table cartAdd = tableEnv.sqlQuery("select \n" +
                "                data['id'] id, \n" +
                "                data['user_id'] user_id, \n" +
                "                data['course_id'] course_id, \n" +
                "                data['course_name'] course_name, \n" +
                "                data['cart_price'] cart_price, \n" +
                "                if(`type`='insert',data['sold'],cast((CAST(data['sold'] AS INT) - CAST(`old`['sold'] AS INT)) as string) \n" +
                "                ) sku_num, \n" +
                "                ts, \n" +
                "                proc_time \n" +
                "            from topic_db where `table`='cart_info' and \n" +
                "                (`type` = 'insert'  or (`type`='update' and `old`['sold'] is not null and (CAST(data['sold'] AS INT) > CAST(`old`['sold'] AS INT))))");
        tableEnv.createTemporaryView("cart_add",cartAdd);
        //tableEnv.executeSql("select * from cart_add").print();

        //TODO 7.将关联的结果写到kafka主题中
        //7.1 创建动态表和要写入的主题进行映射
        tableEnv.executeSql("CREATE TABLE dwd_trade_cart_add (\n" +
                "  id string,\n" +
                "  user_id string,\n" +
                "  course_id string,\n" +
                "  course_name string,\n" +
                "  cart_price string,\n" +
                "  sku_num string,\n" +
                "  ts string,\n" +
                "  proc_time TIMESTAMP_LTZ(3),\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") " + MykafkaUtil.getUpsert("dwd_trade_cart_add"));
        // 写入
        tableEnv.executeSql("insert into dwd_trade_cart_add select * from cart_add");
    }
}
